

package com.hazelcast.instance.impl.executejar;

import com.hazelcast.instance.impl.BootstrappedJetProxy;
import com.hazelcast.instance.impl.executejar.jetservicedecorator.memberside.BootstrapJobDecorator;
import com.hazelcast.jet.JetService;
import com.hazelcast.jet.Job;
import com.hazelcast.jet.config.JobConfig;
import com.hazelcast.jet.core.DAG;
import com.hazelcast.jet.pipeline.Pipeline;

import javax.annotation.Nonnull;

/**
 * This class' state holds {@link ExecuteJobParameters parameters} used by Jet jobs.
 * State is thread-local,so this proxy can be used by multiple threads.
 */
public class MemberJetProxy<M> extends BootstrappedJetProxy<M> {
    private final ThreadLocal<ExecuteJobParameters> executeJobParametersThreadLocal = ThreadLocal.withInitial(ExecuteJobParameters::new);

    public MemberJetProxy(@Nonnull JetService jetService) {
        super(jetService);
    }

    @Override
    public boolean hasExecuteJobParameters() {
        return getExecuteJobParameters() != null;
    }

    @Override
    public ExecuteJobParameters getExecuteJobParameters() {
        return executeJobParametersThreadLocal.get();
    }

    @Override
    public void setExecuteJobParameters(ExecuteJobParameters executeJobParameters) {
        executeJobParametersThreadLocal.set(executeJobParameters);
    }

    @Override
    public void removeExecuteJobParameters() {
        executeJobParametersThreadLocal.remove();
    }

    @Nonnull
    @Override
    public Job newJob(@Nonnull Pipeline pipeline, @Nonnull JobConfig config) {
        Job job = super.newJob(pipeline, config);
        return new BootstrapJobDecorator(job);
    }

    @Nonnull
    @Override
    public Job newJob(@Nonnull DAG dag, @Nonnull JobConfig config) {
        Job job = super.newJob(dag, config);
        return new BootstrapJobDecorator(job);
    }

    @Nonnull
    @Override
    public Job newJobIfAbsent(@Nonnull Pipeline pipeline, @Nonnull JobConfig config) {
        Job job = super.newJobIfAbsent(pipeline, config);
        return new BootstrapJobDecorator(job);
    }

    @Nonnull
    @Override
    public Job newJobIfAbsent(@Nonnull DAG dag, @Nonnull JobConfig config) {
        Job job = super.newJobIfAbsent(dag, config);
        return new BootstrapJobDecorator(job);
    }
}
